<?php

namespace Zeke\Net\Http;


use Zeke\Zee\Event\IoEvent;
use Zeke\Zee\Promise\Promise;
use Zeke\Zee\MiddlewareRunner;
use Zeke\Zee\Event\BufferEvent;
use Zeke\Zee\Event\EventEmitter;
use Zeke\Net\Http\Parser\ChunkedDecoder;
use Zeke\Net\Http\Parser\ChunkedEncoder;
use Psr\Http\Message\ResponseInterface;
use Zeke\Net\Http\Stream\HttpBodyStream;
use Psr\Http\Message\ServerRequestInterface;
use Zeke\Net\Http\Parser\RequestHeaderParser;
use Zeke\Contracts\Io\Socket\ServerContract;
use Zeke\Net\Http\Stream\LengthLimitedStream;
use Zeke\Net\Http\Stream\CloseProtectionStream;
use Zeke\Contracts\Io\Socket\ConnectionContract;
use Zeke\Contracts\Io\Stream\ReadableStreamContracts;
use Zeke\Contracts\Io\Stream\WritableStreamContracts;

/**
 * The `Server` class is responsible for handling incoming connections and then
 * processing each incoming HTTP request.
 *
 * For each request, it executes the callback function passed to the
 * constructor with the respective [request](#request) object and expects
 * a respective [response](#response) object in return.
 *
 * ```php
 * $server = new StreamingServer(function (ServerRequestInterface $request) {
 *     return new Response(
 *         200,
 *         array('Content-Type' => 'text/plain'),
 *         "Hello World!\n"
 *     );
 * });
 * ```
 *
 * In order to process any connections, the server needs to be attached to an
 * instance of `React\Socket\ServerInterface` which emits underlying streaming
 * connections in order to then parse incoming data as HTTP.
 *
 * ```php
 * $socket = new React\Socket\Server(8080, $loop);
 * $server->listen($socket);
 * ```
 *
 * See also the [listen()](#listen) method and the [first example](examples) for more details.
 *
 * When HTTP/1.1 clients want to send a bigger request body, they MAY send only
 * the request headers with an additional `Expect: 100-continue` header and
 * wait before sending the actual (large) message body.
 * In this case the server will automatically send an intermediary
 * `HTTP/1.1 100 Continue` response to the client.
 * This ensures you will receive the request body without a delay as expected.
 * The [Response](#response) still needs to be created as described in the
 * examples above.
 *
 * See also [request](#request) and [response](#response)
 * for more details (e.g. the request data body).
 *
 * The `Server` supports both HTTP/1.1 and HTTP/1.0 request messages.
 * If a client sends an invalid request message, uses an invalid HTTP protocol
 * version or sends an invalid `Transfer-Encoding` in the request header, it will
 * emit an `error` event, send an HTTP error response to the client and
 * close the connection:
 *
 * ```php
 * $server->on('error', function (Exception $e) {
 *     echo 'Error: ' . $e->getMessage() . PHP_EOL;
 * });
 * ```
 *
 * Note that the request object can also emit an error.
 * Check out [request](#request) for more details.
 *
 * @see Request
 * @see Response
 * @see self::listen()
 */
final class StreamingServer extends EventEmitter
{


    private $callback;

    /**
     * Creates an HTTP server that invokes the given callback for each incoming HTTP request
     *
     * In order to process any connections, the server needs to be attached to an
     * instance of `React\Socket\ServerInterface` which emits underlying streaming
     * connections in order to then parse incoming data as HTTP.
     * See also [listen()](#listen) for more details.
     *
     * @param callable|callable[] $requestHandler
     * @see self::listen()
     */
    public function __construct($requestHandler)
    {
        if (!is_callable($requestHandler) && !is_array($requestHandler)) {
            throw new \InvalidArgumentException('Invalid request handler given');
        } elseif (!is_callable($requestHandler)) {
            $requestHandler = new MiddlewareRunner($requestHandler);
        }

        // 定义server时传入的回调，和一些预定义的中间件（把对象当方法执行的那种）
        // MiddlewareRunner 在执行的时候,会先获取下一个执行的中间件，然后执行当前的中间件，返回下一个中间件
        $this->callback = $requestHandler;
    }

    /**
     * Starts listening for HTTP requests on the given socket server instance
     *
     * The server needs to be attached to an instance of
     * `React\Socket\ServerInterface` which emits underlying streaming
     * connections in order to then parse incoming data as HTTP.
     * For each request, it executes the callback function passed to the
     * constructor with the respective [request](#request) object and expects
     * a respective [response](#response) object in return.
     *
     * You can attach this to a
     * [`React\Socket\Server`](https://github.com/reactphp/socket#server)
     * in order to start a plaintext HTTP server like this:
     *
     * ```php
     * $server = new StreamingServer($handler);
     *
     * $socket = new React\Socket\Server(8080, $loop);
     * $server->listen($socket);
     * ```
     *
     * See also [example #1](examples) for more details.
     *
     * Similarly, you can also attach this to a
     * [`React\Socket\SecureServer`](https://github.com/reactphp/socket#secureserver)
     * in order to start a secure HTTPS server like this:
     *
     * ```php
     * $server = new StreamingServer($handler);
     *
     * $socket = new React\Socket\Server(8080, $loop);
     * $socket = new React\Socket\SecureServer($socket, $loop, array(
     *     'local_cert' => __DIR__ . '/localhost.pem'
     * ));
     *
     * $server->listen($socket);
     * ```
     *
     * See also [example #11](examples) for more details.
     *
     * @param ServerContract $socketServer
     */
    public function listen(ServerContract $socketServer)
    {
        $socketServer->on(IoEvent::CONNECT, array($this, 'handleConnection'));
    }


    /**
     * IoEvent::CONNECT事件触发后执行的回调
     *
     * @param ConnectionContract $conn
     * @see StreamingServer::listen()
     * @author <liu.bin@ruaby.com> 2018-08-21 23:19
     */
    public function handleConnection(ConnectionContract $conn)
    {
        $that = $this;

        // 获取本机监听地址
        $uriLocal = $conn->getLocalAddress();
        if ($uriLocal !== null) {
            // local URI known, so translate transport scheme to application scheme
            // $uriLocal tcp://转换成http://，tls://转换成https://
            $uriLocal = strtr($uriLocal, array('tcp://' => 'http://', 'tls://' => 'https://'));
        }

        //header解析类
        $parser = new RequestHeaderParser($uriLocal, $conn->getRemoteAddress());
        $listenerHandler = array($parser, 'feed');


        //注册接收到客户端消息后的处理函数
        $conn->on(BufferEvent::DATA, $listenerHandler);


        //当$parser接收到完整的header后，会格式化成$request对象，并执行此回调
        $parser->on('headers', function (ServerRequestInterface $request, $bodyBuffer) use ($conn, $listenerHandler, $that) {

            // parsing request completed => stop feeding parser
            // 此处移除监听事件函数，在handleRequest中，conn会被重新封装成httpBodyStream,
            // 具体的BufferEvent::DATA事件，由httpBodyStream处理，主要获取body数据内容
            $conn->removeListener(BufferEvent::DATA, $listenerHandler);

            // 处理请求:
            // 1:http header 语法验证,传输协议解析
            // 2:执行定义好的中间件和回调函数，http回调函数最后一个调用（业务逻辑，用户定义）,
            // 3:用户回调中需要返回response对象（$callback返回），向客户端输出
            $that->handleRequest($conn, $request);


            // 此处的$bodyBuffer，是切割header后剩余的数据
            if ($bodyBuffer !== '') {
                // 手动触发body数据
                // 此处会逐级向上调用httpBodyStream的BufferEvent::DATA事件处理函数
                $conn->emit(BufferEvent::DATA, array($bodyBuffer));
            }
        });


        //注册header解析错误的回调处理函数
        $parser->on('error', function(\Exception $e) use ($conn, $listenerHandler, $that) {
            $conn->removeListener('data', $listenerHandler);
            $that->emit('error', array($e));
            $that->writeError($conn, $e->getCode() !== 0 ? $e->getCode() : 400);
        });



    }


    /**
     * header解析成功后，执行此函数
     *
     * @param ConnectionContract $conn
     * @param ServerRequestInterface $request header解析成功后，生成$request对象
     * @return mixed
     * @author <liu.bin@ruaby.com> 2018-08-21 23:31
     */
    public function handleRequest(ConnectionContract $conn, ServerRequestInterface $request)
    {
        // 封装数据流
        $parseStream = $this->_parseStream($conn,$request);
        if(false === $parseStream){
            return false;
        }

        list($stream,$contentLength) = $parseStream;

        //HttpBodyStream实现了StreamInterface接口，知道了$contentLength的长度
        //具体的body信息将会在 RequestBodyBufferMiddleware 中间件中根据HttpBodyStream读取
        $request = $request->withBody(new HttpBodyStream($stream, $contentLength));

        // 客户端需要发送body数据，但是在header中设置了Expect:100-continue,询问服务器是否支持body数据
        // 服务器需要响应HTTP/1.1 100 Continue\r\n\r\n，表明支持
        if ($request->getProtocolVersion() !== '1.0' && '100-continue' === strtolower($request->getHeaderLine('Expect'))) {
            $conn->write("HTTP/1.1 100 Continue\r\n\r\n");
        }


        //处理响应
        $this->_parseResponse($request,$conn,$stream,$contentLength);

        return true;
    }


    /**
     * 解析流处理方式
     *
     * @param $request
     * @param $stream
     * @return array|bool
     *
     * @author <liu.bin@ruaby.com> 2018-08-29 23:23
     */
    private function _parseStream(ConnectionContract $conn, ServerRequestInterface $request)
    {

        // CloseProtectionStream 在调用close的时候，不会调用fclose()关闭socket资源，只是移除事件侦听
        $stream = new CloseProtectionStream($conn);
        $contentLength = 0;
        if ($request->hasHeader('Transfer-Encoding')) {
            // Transfer-Encoding传输编码

            //Transfer-Encoding 只支持chunked
            if (strtolower($request->getHeaderLine('Transfer-Encoding')) !== 'chunked') {
                $this->emit('error', array(new \InvalidArgumentException('Only chunked-encoding is allowed for Transfer-Encoding')));
                $this->writeError($conn, 501, $request);
                return false;
            }

            // Transfer-Encoding: chunked and Content-Length header MUST NOT be used at the same time
            // as per https://tools.ietf.org/html/rfc7230#section-3.3.3
            // Transfer-Encoding和Content-Length不可以同时使用
            if ($request->hasHeader('Content-Length')) {
                $this->emit('error', array(new \InvalidArgumentException('Using both `Transfer-Encoding: chunked` and `Content-Length` is not allowed')));
                $this->writeError($conn, 400, $request);
                return false;
            }

            $stream = new ChunkedDecoder($stream);
            $contentLength = null;
        } elseif ($request->hasHeader('Content-Length')) {
            //设置Content-Length（body体长度），根据此值，获取http完整数据

            $string         = $request->getHeaderLine('Content-Length');
            $contentLength  = (int)$string;
            if ((string)$contentLength !== $string) {
                // Content-Length value is not an integer or not a single integer
                $this->emit('error', array(new \InvalidArgumentException('The value of `Content-Length` is not valid')));
                $this->writeError($conn, 400, $request);
                return false;
            }

            //定义获取指定长度的stream对象
            //LengthLimitedStream会在读取的data长度达到$contentLength后，触发data事件
            $stream = new LengthLimitedStream($stream, $contentLength);
        }
        return [$stream,$contentLength];
    }


    /**
     * 处理
     *
     * @param ServerRequestInterface $request
     * @param ConnectionContract $conn
     * @param ReadableStreamContracts $stream
     * @param $contentLength
     *
     * @author <liu.bin@ruaby.com> 2018-08-29 23:14
     */
    private function _parseResponse(ServerRequestInterface $request,ConnectionContract $conn,ReadableStreamContracts $stream,$contentLength)
    {

        // execute request handler callback
        // $callback = MiddlewareRunner:默认的中间件＋请求处理回调函数集合
        // 1:LimitConcurrentRequestsMiddleware中间件：当前最大连接数？原代码中返回的是一个promise
        // 2:RequestBodyBufferMiddleware中间件：获取http的body数据：原代码中返回的是一个promise
        // 3:RequestBodyParserMiddleware中间件：将body转换成$_POST
        $callback = $this->callback;
        try {
            //执行中间件：执行后，将返回最后一个中间件的返回值，一般为httpServer的业务回调函数返回值$response

            //同步模式
            //此处执行第一个中间件invoke函数，处理完自己的业务后会执行第二个中间件invoke函数，类推...,
            //最终返回最后一个中间件的返回值，即为response

            //异步模式
            //执行第一个中间件invoke函数，并不会立即执行第二个中间件的invoke，而是直接返回一个promise;
            //promise里面注册了调用第二个中间invoke的回调函数，等待触发（触发由promise内部发起）
            $response = $callback($request);
        } catch (\Exception $error) {
            // request handler callback throws an Exception
            $response = new Response(200,[],'error-1');
        } catch (\Throwable $error) { // @codeCoverageIgnoreStart
            // request handler callback throws a PHP7+ Error
            $response = new Response(200,[],'error-2');
        }

        // happy path: request body is known to be empty => immediately end stream
        // 如果是null,则是ChunkedDecoder
        // 如果是数字,则是LengthLimitedStream
        if ($contentLength === 0) {
            $stream->emit('end');
            $stream->close();
        }

        // happy path: response returned, handle and return immediately
        if ($response instanceof ResponseInterface) {
            $this->handleResponse($conn, $request, $response);
        }

        // 第一次中间件是 RequestBodyBufferMiddleware
        // 首选返回一个promise对象 $response == promise
        // $response 注册$resolve回调函数，在RequestBodyBufferMiddleware读取完整body后调用
        // $stack == RequestBodyParserMiddleware,执行完自己的业务逻辑后，同步调用http的回调函数
        // 返回$response == ResponseInterface
        // 如果都是异步执行，则需要在中间内部获取下一级中间件（$next()调用返回）返回的promise，注册then方法
        if($response instanceof Promise){
            $response->then(function($param) use($conn){
                list($stack,$request) = $param;
                $response = $stack($request);
                if($response instanceof ResponseInterface){
                    $this->handleResponse($conn, $request, $response);
                }
            });

        }


    }

    /**
     * 按照http格式输出数据
     *
     * @param ConnectionContract $connection
     * @param ServerRequestInterface $request
     * @param ResponseInterface $response
     *
     * @author <liu.bin@ruaby.com> 2018-08-22 15:17
     */
    public function handleResponse(ConnectionContract $connection, ServerRequestInterface $request, ResponseInterface $response)
    {

        // return early and close response body if connection is already closed
        $body = $response->getBody();
        if (!$connection->isWritable()) {
            $body->close();
            return;
        }

        $response = $response->withProtocolVersion($request->getProtocolVersion());

        // assign default "X-Powered-By" header as first for history reasons
        if (!$response->hasHeader('X-Powered-By')) {
            $response = $response->withHeader('X-Powered-By', 'ZekePHP/alpha');
        }

        if ($response->hasHeader('X-Powered-By') && $response->getHeaderLine('X-Powered-By') === ''){
            $response = $response->withoutHeader('X-Powered-By');
        }

        $response = $response->withoutHeader('Transfer-Encoding');

        // assign date header if no 'date' is given, use the current time where this code is running
        if (!$response->hasHeader('Date')) {
            // IMF-fixdate  = day-name "," SP date1 SP time-of-day SP GMT
            $response = $response->withHeader('Date', gmdate('D, d M Y H:i:s') . ' GMT');
        }

        if ($response->hasHeader('Date') && $response->getHeaderLine('Date') === ''){
            $response = $response->withoutHeader('Date');
        }

        if (!$body instanceof HttpBodyStream) {
            $response = $response->withHeader('Content-Length', (string)$body->getSize());
        } elseif (!$response->hasHeader('Content-Length') && $request->getProtocolVersion() === '1.1') {
            // assign chunked transfer-encoding if no 'content-length' is given for HTTP/1.1 responses
            $response = $response->withHeader('Transfer-Encoding', 'chunked');
        }

        // HTTP/1.1 assumes persistent connection support by default
        // we do not support persistent connections, so let the client know
        // HTTP/1.1 默认支持持久化连接
        // 目前版本不支持，所以需要通知客户端
        if ($request->getProtocolVersion() === '1.1') {
            $response = $response->withHeader('Connection', 'close');
        }
        // 2xx response to CONNECT and 1xx and 204 MUST NOT include Content-Length or Transfer-Encoding header
        //1:如果请求类型为CONNECT,且正常的情况下，不需要返回body;
        //2:临时响应1xx,不需要返回body;
        //3:204，此状态定义为没有body;
        $code = $response->getStatusCode();
        if (($request->getMethod() === 'CONNECT' && $code >= 200 && $code < 300) || ($code >= 100 && $code < 200) || $code === 204) {
            $response = $response->withoutHeader('Content-Length')->withoutHeader('Transfer-Encoding');
        }

        // 101 (Switching Protocols) response uses Connection: upgrade header
        // persistent connections are currently not supported, so do not use
        // this for any other replies in order to preserve "Connection: close"
        // 101 切换协议，响应客户端 upgrade
        if ($code === 101) {
            $response = $response->withHeader('Connection', 'upgrade');
        }

        // 101 (Switching Protocols) response (for Upgrade request) forwards upgraded data through duplex stream
        // 2xx (Successful) response to CONNECT forwards tunneled application data through duplex stream
        if (($code === 101 || ($request->getMethod() === 'CONNECT' && $code >= 200 && $code < 300)) && $body instanceof HttpBodyStream && $body->input instanceof WritableStreamContracts) {
            if ($request->getBody()->isReadable()) {
                // request is still streaming => wait for request close before forwarding following data from connection
                $request->getBody()->on('close', function () use ($connection, $body) {
                    if ($body->input->isWritable()) {
                        $connection->pipe($body->input);
                        $connection->resume();
                    }
                });
            } elseif ($body->input->isWritable()) {
                // request already closed => forward following data from connection
                $connection->pipe($body->input);
                $connection->resume();
            }
        }

        // build HTTP response header by appending status line and header fields
        $headers = "HTTP/" . $response->getProtocolVersion() . " " . $response->getStatusCode() . " " . $response->getReasonPhrase() . "\r\n";
        foreach ($response->getHeaders() as $name => $values) {
            foreach ($values as $value) {
                $headers .= $name . ": " . $value . "\r\n";
            }
        }

        // response to HEAD and 1xx, 204 and 304 responses MUST NOT include a body
        // exclude status 101 (Switching Protocols) here for Upgrade request handling above
        if ($request->getMethod() === 'HEAD' || $code === 100 || ($code > 101 && $code < 200) || $code === 204 || $code === 304) {
            $body = '';
        }

        // this is a non-streaming response body or the body stream already closed?
        if (!$body instanceof ReadableStreamContracts || !$body->isReadable()) {
            // add final chunk if a streaming body is already closed and uses `Transfer-Encoding: chunked`
            if ($body instanceof ReadableStreamContracts && $response->getHeaderLine('Transfer-Encoding') === 'chunked') {
                $body = "0\r\n\r\n";
            }

            // end connection after writing response headers and body
            $connection->write($headers . "\r\n" . $body);
            $connection->end();
            return;
        }

        $connection->write($headers . "\r\n");

        if ($response->getHeaderLine('Transfer-Encoding') === 'chunked') {
            $body = new ChunkedEncoder($body);
        }

        // Close response stream once connection closes.
        // Note that this TCP/IP close detection may take some time,
        // in particular this may only fire on a later read/write attempt.
        //注册close监听事件
        $connection->on('close', array($body, 'close'));

        $body->pipe($connection);
    }



    /**
     * 向客户端返回http错误信息
     *
     * @param ConnectionContract $conn
     * @param $code
     * @param ServerRequestInterface|null $request
     *
     * @author <liu.bin@ruaby.com> 2018-08-22 15:15
     */
    public function writeError(ConnectionContract $conn, $code, ServerRequestInterface $request = null)
    {

        $response = new Response($code, array('Content-Type' => 'text/plain'), 'Error ' . $code);

        // append reason phrase to response body if known
        $reason = $response->getReasonPhrase();
        if ($reason !== '') {
            $body = $response->getBody();
            $body->seek(0, SEEK_END);
            $body->write(': ' . $reason);
        }

        if ($request === null) {
            $request = new Request('GET', '/', array(), null, '1.1');
        }

        $this->handleResponse($conn, $request, $response);
    }
}
